#include "XProducerConsumer.h"


namespace Xts
{
	namespace Util
	{
		XProducerConsumer::XProducerConsumer(void)
			:x_pJobQueue(new XRequestQueue()), x_bIsRunning(false)
		{
		}

		XProducerConsumer::~XProducerConsumer(void)
		{
			if (IsRunning())
				StopReaderThreads();
			x_pJobQueue->Clear();
		}

		void XProducerConsumer::StartReaderThreads(unsigned int n)
		{
			if (IsRunning())
				return;
			for (unsigned int i = 0; i < n; i++)
			{
				::IceUtil::ThreadPtr t = new XRequestReaderThread(x_pJobQueue);
				x_JobReaderThreadCtrlSet.push_back(t->start());
			}
			x_bIsRunning = true;
		}

		void XProducerConsumer::StopReaderThreads()
		{
			if (!IsRunning())
				return;

			for (size_t i = 0; i < x_JobReaderThreadCtrlSet.size(); i++)
			{
				IXRequestJobPtr pEndJob = new XEndingReaderJob();
				x_pJobQueue->Enqueue(pEndJob);
			}

			::std::vector<::IceUtil::ThreadControl>::iterator iter;
			for (iter = x_JobReaderThreadCtrlSet.begin(); iter != x_JobReaderThreadCtrlSet.end(); iter++)
			{
				iter->join();
			}

			x_JobReaderThreadCtrlSet.clear();
			x_bIsRunning = false;
		}

		void XProducerConsumer::NewRequestJob(const IXRequestJobPtr& pJob)
		{
			x_pJobQueue->Enqueue(pJob);
		}
	}
}


